package db;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import offline.CGA;
import redis.clients.jedis.Response;
import tlsb.DB;
import tlsb.Utils;
import util.IPartition;
import util.RedisCluster;
import util.RedisClusterFactory;

/**
 * 目前只有所有vip的用户的status采用的是pull的方式获取的 status采用的按照uid进行parition的
 * SETOFFOLLOWERS与uid存在同一台机器上？ SETOFFOLLOWEES与uid存在同一台机器上？ SETOFV 存在metadata
 * redis上
 * 
 * @author weijinxian
 * @modify xiafan 2012-11-03
 * 
 */

public class RedisClientHYBRID extends DB {
	static AtomicLong midCounter = new AtomicLong(0);
	final private long STARTOFFSET = 0;
	final private long ENDOFFSET = 19;
	final private static int MAX = 999999999;
	final private String PREFIXOFUTL = "hu:";
	final private String SETOFV = "vset";
	final private int HTLLENGTH = 20;
	private static final String SETOFFOLLOWERS = ":flr";
	private static final String SETOFFOLLOWEES = ":fle";
	final private String PREFIXOFHID = "hh:";
	// final private String UIDOFMID = ""
	static Random random = new Random();
	private boolean _Debug = false;

	private RedisCluster folRedisCluster = RedisClusterFactory.instance
			.getFolRedisCluster();
	private RedisCluster statusRedisCluster = RedisClusterFactory.instance
			.getStatusRedisCluster();
	private RedisCluster metaDataRedisCluster = RedisClusterFactory.instance
			.getMetaDataRedisCluster();

	public RedisClientHYBRID() {

	}

	@Override
	public void cleanup() {
		folRedisCluster.close();
		statusRedisCluster.close();
		metaDataRedisCluster.close();
	}

	@Override
	public int follow(String uid, String fid) {
		// TODO Auto-generated method stub
		// add to new followers and followers for different strateges; also
		// update owns(first) and belongs
		return 0;
	}

	@Override
	// TODO uid change string into integer.
	public int getHTL(String uid, int start) {

		// 读取已经push的status
		List<String> ptl = statusRedisCluster.getJedis(
				statusRedisCluster.getPartition().shard(uid)).lrange(
				Utils.joinStr(PREFIXOFHID, uid), STARTOFFSET, ENDOFFSET);

		// pull the timeline of vips
		Set<String> curVset = new HashSet<String>();
		for (String mem : folRedisCluster.getJedis(
				folRedisCluster.getPartition().shard(uid)).smembers(
				Utils.joinStr(SETOFFOLLOWEES, uid))) {
			if (metaDataRedisCluster.getJedis(
					metaDataRedisCluster.getPartition().shard(SETOFV))
					.sismember(SETOFV, mem)) {
				curVset.add(mem);
			}
		}
		List<List<String>> tls = new LinkedList<List<String>>();
		tls.add(ptl);
		List<Response<List<String>>> responses = new LinkedList<Response<List<String>>>();
		if (curVset != null) {
			for (String vmem : curVset) {
				int nid = statusRedisCluster.getPartition().shard(vmem);
				responses.add(statusRedisCluster.getPipeline(nid).lrange(
						Utils.joinStr(PREFIXOFUTL, vmem), STARTOFFSET,
						ENDOFFSET));
			}
			statusRedisCluster.synAllPipelines();
			for (Response<List<String>> response : responses) {
				tls.add(response.get());
			}
		}
		if (tls != null) {
			mergeFilter(tls, HTLLENGTH, start);
		}
		if (_Debug) {
			// System.out.println(uid + " OVLP:" + htl);
		}
		return 0;
	}

	public static String[] mergeFilter(List<List<String>> stls, int count,
			int start) {
		String[] htl = new String[count];
		String nextMid = null;
		String lastMid = "-1";
		String thisMid = "";
		List<String> thisStl = new LinkedList<String>();
		int stlNo = MAX;
		for (int i = 0; i < count + start; i++) {

			for (int j = 0; j < stls.size(); j++) {
				// check
				thisStl = stls.get(j);
				if (thisStl.size() == 0)
					continue;
				thisMid = thisStl.get(thisStl.size() - 1);
				// remove duplicates
				if (thisMid.equals(lastMid)) {
					thisStl.remove(thisStl.size() - 1);
					if (thisStl.size() == 0)
						continue;
					thisMid = thisStl.get(thisStl.size() - 1);
				}

				if (nextMid == null
						|| Long.parseLong(nextMid) < Long.parseLong(thisMid)) {
					nextMid = thisMid;
					stlNo = j;
				}
			}
			if (stlNo == MAX)
				break;
			if (i >= start) {
				htl[i] = nextMid;
			}
			stls.get(stlNo).remove(stls.get(stlNo).size() - 1);
			lastMid = nextMid;
			nextMid = null;
			stlNo = MAX;
		}

		return htl;
	}

	@Override
	public int newUser() {
		// TODO Auto-generated method stub
		return 0;
	}

	public static void main(String[] args) {
		List<List<String>> teststls = new LinkedList<List<String>>();

		for (int i = 0; i < 5; i++) {
			LinkedList<String> stl = new LinkedList();
			for (int j = i; j < 5; j++) {
				stl.add(j + "");
			}
			teststls.add(stl);
			System.out.println(i + "~~~" + stl.toString());
		}
		RedisClientHYBRID rch = new RedisClientHYBRID();

		for (String s : rch.mergeFilter(teststls, 20, 0)) {
			System.out.println(s);
		}
	}

	@Override
	public int post(String uid, String mid) {
		// judeg wether current user is a VIP user
		if (!(metaDataRedisCluster.getJedis(metaDataRedisCluster.getPartition()
				.shard(SETOFV)).sismember(SETOFV, uid))) {
			for (String f : folRedisCluster.getJedis(
					folRedisCluster.getPartition().shard(uid)).smembers(
					Utils.joinStr(uid, SETOFFOLLOWERS))) {
				int nid = statusRedisCluster.getPartition().shard(f);
				statusRedisCluster.getPipeline(nid).lpush(
						Utils.joinStr(PREFIXOFHID, f), mid);
				midCounter.incrementAndGet();
				if (midCounter.get() % 1000000 == 0)
					System.err.println(midCounter.get());
			}
			statusRedisCluster.synAllPipelines();
		} else {
			statusRedisCluster.getJedis(
					statusRedisCluster.getPartition().shard(uid)).lpush(
					Utils.joinStr(PREFIXOFUTL, uid), mid);
		}
		return 0;
	}

	@Override
	public int unfollow(String uid, String fid) {
		// TODO Auto-generated method stub
		return 0;
	}

}
